package com.cmdi.mq.demo1;

import io.netty.channel.DefaultChannelId;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @Description: 有序发送消息
 * @Author: gaozb
 */
public class ProducerOrdered {
    public static void main(String[] args) throws Exception {
        //指定生产者组
        DefaultMQProducer producer = new DefaultMQProducer("producer-group");
        // 设置NameServer的地址
        producer.setNamesrvAddr("10.254.243.246:9876;10.254.243.249:9876");
        //解决invokeSync call timeout异常
        DefaultChannelId.newInstance();
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 5; i++) {
            for(int j = 0; j < 10; j++) {
                //向topic1发送消息。如果没有topic1需要首先创建
                Message message = new Message("topic1", "tagA", ("HelloRocketMQ->" + i + ":" + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
                System.out.println(new String(message.getBody()));
                //同步发送消息，等待发送结果
                SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                        //list就是当前所有的queue，message就是要发送的消息，o就是send传入的第三个参数，此处是i
                        Integer id = (Integer) o;
                        //通过i决定了当前消息发送到哪个queue中，因此相同i的肯定发送到一个queue中
                        int index = id % list.size();
                        return list.get(index);
                    }
                }, i);
                System.out.println(sendResult);
            }
        }
        //消息发送完成后，关闭producer服务
        producer.shutdown();
    }
}